package report;

import com.alibaba.fastjson.JSON;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.apache.jmeter.engine.util.NoThreadClone;
import org.apache.jmeter.samplers.Remoteable;
import org.apache.jmeter.samplers.SampleEvent;
import org.apache.jmeter.samplers.SampleListener;
import org.apache.jmeter.samplers.SampleResult;
import org.apache.jmeter.services.FileServer;
import org.apache.jmeter.testelement.AbstractTestElement;
import org.apache.jmeter.testelement.TestStateListener;
import org.apache.jmeter.testelement.ThreadListener;
import org.apache.jmeter.util.JMeterUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import report.entity.MySampleResult;

import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Properties;
import java.util.Random;

/**
 * 单线程的生产者 优点是确保写入的消息是顺序的
 */
public class MySummariser extends AbstractTestElement
		implements Serializable, SampleListener, TestStateListener, Remoteable, ThreadListener, NoThreadClone {

	private static final Logger log = LoggerFactory.getLogger(MySummariser.class);

	private static final long serialVersionUID = 1995L;

	private static final String KAFKA_HOST = JMeterUtils.getPropDefault("KAFKA_HOST", "127.0.0.1:9092");
	private static final String KAFKA_TOPIC = JMeterUtils.getPropDefault("KAFKA_TOPIC", "perf_report_topic");
	private static final String WEB_SERVER = JMeterUtils.getPropDefault("WEB_SERVER",
			"http://127.0.0.1:18080/perf/report/finish?uuid=");

	private static String currentHost;

	static {
		try {
			currentHost = InetAddress.getLocalHost().getHostAddress();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
	}

	private String reportId = FileServer.getFileServer().getScriptName().replace(".jmx", "").replace("perf_report_", "");
	private Properties props = new Properties();
	private ArrayList<KafkaProducer<String, String>> producerList = new ArrayList<>();
	private Random random = new Random();

	{
		props.put("bootstrap.servers", KAFKA_HOST);
		props.put("acks", "0");
		props.put("retries", "0");
		props.put("batch.size", "16384");
		props.put("linger.ms", "1");
		props.put("buffer.memory", "33554432");
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
	}
	public MySummariser() {
		super();
	}

	public MySummariser(String name) {
		this();
		setName(name);
	}

	public static void main(String[] args) {
		System.out.println(JSON
				.toJSONString(new MySampleResult(currentHost, 10, 10, 10, 10, 10, "reportId", "threadGroupName", 100)));
	}

	@Override
	public void threadFinished() {

	}

	@Override
	public void threadStarted() {

	}

	@Override
	public void testEnded() {
		producerList.forEach(KafkaProducer::close);
		CloseableHttpClient httpclient = HttpClients.createDefault();
		HttpGet get = new HttpGet(WEB_SERVER + reportId);
		try {
			String response = EntityUtils.toString(httpclient.execute(get).getEntity());
			log.info("finish消息发送结果：{}", response);
			httpclient.close();
		} catch (Exception e) {
			e.printStackTrace();
			log.error("finish消息发送失败：{}", e.getMessage());
		}
	}

	@Override
	public void testEnded(String string) {
		testEnded();
	}

	@Override
	public void testStarted() {
		for (int i = 0; i < 20; i++) {
			producerList.add(new KafkaProducer<>(props));
			log.info("创建生产者OK：{}", i);
		}
	}

	@Override
	public void testStarted(String string) {
		testStarted();
	}

	@Override
	public void sampleOccurred(SampleEvent event) {

		SampleResult result = event.getResult();

		// 收集采样结果
		MySampleResult sampleResult;
		if (result.isSuccessful()) {
			sampleResult = new MySampleResult(currentHost, result.getSampleCount(), result.getErrorCount(),
					result.getSentBytes(), result.getBytesAsLong(), result.getAllThreads(),
					result.getSampleLabel(), result.getThreadName(), result.getTime());
		} else {
			sampleResult = new MySampleResult(currentHost, result.getSampleCount(), result.getErrorCount(),
					result.getSentBytes(), result.getBytesAsLong(), result.getAllThreads(),
					result.getSampleLabel(), result.getThreadName(), result.getTime(), result.getSamplerData(),
					result.getResponseDataAsString());
		}

		// 发送消息
		producerList.get(random.nextInt(producerList.size())).send(new ProducerRecord<>(KAFKA_TOPIC, reportId, JSON.toJSONString(sampleResult)));

		log.debug("消息发送到kafka成功");
	}

	@Override
	public void sampleStarted(SampleEvent event) {

	}

	@Override
	public void sampleStopped(SampleEvent event) {

	}
}
